package com.diy.sigmund.diycommon.test.controller.redis;

import com.diy.sigmund.diycommon.entity.ResultMsg;
import com.diy.sigmund.diycommon.middleware.cache.redis.RedissonConfig;
import com.diy.sigmund.diycommon.test.entity.AnyObject;
import com.diy.sigmund.diycommon.test.entity.RedissonDTO;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.redisson.api.BatchOptions;
import org.redisson.api.BatchOptions.ExecutionMode;
import org.redisson.api.BatchResult;
import org.redisson.api.GeoEntry;
import org.redisson.api.GeoPosition;
import org.redisson.api.GeoUnit;
import org.redisson.api.RAtomicDouble;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBatch;
import org.redisson.api.RBinaryStream;
import org.redisson.api.RBitSet;
import org.redisson.api.RBucket;
import org.redisson.api.RBuckets;
import org.redisson.api.RFuture;
import org.redisson.api.RGeo;
import org.redisson.api.RKeys;
import org.redisson.api.RList;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RPermitExpirableSemaphore;
import org.redisson.api.RScript;
import org.redisson.api.RScript.Mode;
import org.redisson.api.RSemaphore;
import org.redisson.api.RSet;
import org.redisson.api.RTransaction;
import org.redisson.api.RedissonClient;
import org.redisson.api.TransactionOptions;
import org.redisson.client.RedisClient;
import org.redisson.client.RedisClientConfig;
import org.redisson.client.RedisConnection;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.transaction.TransactionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.interceptor.TransactionAspectSupport;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author ylm-sigmund
 * @since 2021/11/22 10:38
 */
@RestController
public class RedissonController<V> {

    private static final Logger log = LoggerFactory.getLogger(RedissonController.class);
    // @Autowired
    // private RedissonTransactionManager customRedissonTxManager;

    @PostMapping("/redisson/testGetKeys")
    public <V> ResultMsg testGetKeys(@RequestBody RedissonDTO redissonDTO) throws IOException {
        RedissonClient redisson = RedissonConfig.getRedisson();
        // 6. 分布式对象
        // 每个Redisson对象实例都会有一个与之对应的Redis数据实例，可以通过调用getName方法来取得Redis数据实例的名称（key）
        RMap map = redisson.getMap("mymap");
        // mymap
        String name = map.getName();
        log.info("name:{}", name);

        // 所有与Redis key相关的操作都归纳在RKeys这个接口里
        RKeys keys = redisson.getKeys();

        Iterable<String> allKeys = keys.getKeys();
        Iterable<String> foundedKeys = keys.getKeysByPattern("key*");
        long numOfDeletedKeys = keys.delete("obj1", "obj2", "obj3");
        long deletedKeysAmount = keys.deleteByPattern("test?");
        String randomKey = keys.randomKey();
        long keysAmount = keys.count();

        // 6.1. 通用对象桶（Object Bucket）
        // Redisson的分布式RBucketJava对象是一种通用对象桶可以用来存放任类型的对象。 除了同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

        RBucket<AnyObject> bucket = redisson.getBucket("anyObject");
        bucket.set(new AnyObject(1));
        AnyObject obj = bucket.get();

        bucket.trySet(new AnyObject(3));
        bucket.compareAndSet(new AnyObject(4), new AnyObject(5));
        bucket.getAndSet(new AnyObject(6));
        // 还可以通过RBuckets接口实现批量操作多个RBucket对象：

        RBuckets buckets = redisson.getBuckets();
        // List<RBucket<V>> foundBuckets = buckets.find("myBucket*");
        Map<String, V> loadedBuckets = buckets.get("myBucket1", "myBucket2", "myBucket3");

        Map<String, Object> map1 = new HashMap<>();
        map1.put("myBucket1", new AnyObject());
        map1.put("myBucket2", new AnyObject());

        // 利用Redis的事务特性，同时保存所有的通用对象桶，如果任意一个通用对象桶已经存在则放弃保存其他所有数据。
        buckets.trySet(map1);
        // 同时保存全部通用对象桶。
        buckets.set(map1);

        // 6.2. 二进制流（Binary Stream）
        // Redisson的分布式RBinaryStream Java对象同时提供了InputStream接口和OutputStream接口的实现。流的最大容量受Redis主节点的内存大小限制。

        RBinaryStream stream = redisson.getBinaryStream("anyStream");
        byte[] content = "可爱的羊巴鲁，刚好拿下印度，就把你给毙啦".getBytes();
        stream.set(content);

        InputStream is = stream.getInputStream();
        byte[] readBuffer = new byte[512];
        is.read(readBuffer);

        OutputStream os = stream.getOutputStream();
        byte[] contentToWrite = "可爱的羊巴鲁，刚好拿下印度，就把你给毙啦".getBytes();
        os.write(contentToWrite);

        // 6.3. 地理空间对象桶（Geospatial Bucket）
        // Redisson的分布式RGeo Java对象是一种专门用来储存与地理位置有关的对象桶。除了同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

        RGeo<String> geo = redisson.getGeo("test");
        geo.add(new GeoEntry(13.361389, 38.115556, "Palermo"),
                new GeoEntry(15.087269, 37.502669, "Catania"));
        geo.addAsync(37.618423, 55.751244, "Moscow");

        Double distance = geo.dist("Palermo", "Catania", GeoUnit.METERS);
        geo.hashAsync("Palermo", "Catania");
        Map<String, GeoPosition> positions = geo.pos("test2", "Palermo", "test3", "Catania", "test1");
        List<String> cities = geo.radius(15, 37, 200, GeoUnit.KILOMETERS);
        Map<String, GeoPosition> citiesWithPositions = geo.radiusWithPosition(15, 37, 200, GeoUnit.KILOMETERS);

        // 6.4. BitSet
        // Redisson的分布式RBitSetJava对象采用了与java.util.BiteSet类似结构的设计风格。可以理解为它是一个分布式的可伸缩式位向量。需要注意的是RBitSet的大小受Redis限制，最大长度为4 294 967 295。除了同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

        RBitSet set = redisson.getBitSet("simpleBitset");
        set.set(0, true);
        set.set(1812, false);
        set.clear(0);
        // set.addAsync("e");
        set.andAsync("e");
        set.xor("anotherBitset");

        // 6.5. 原子整长形（AtomicLong）
        // Redisson的分布式整长形RAtomicLong对象和Java中的java.util.concurrent.atomic.AtomicLong对象类似。除了同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

        RAtomicLong atomicLong = redisson.getAtomicLong("myAtomicLong");
        atomicLong.set(3);
        atomicLong.incrementAndGet();
        atomicLong.get();

        // 6.6. 原子双精度浮点（AtomicDouble）
        // Redisson还提供了分布式原子双精度浮点RAtomicDouble，弥补了Java自身的不足。除了同步接口外，还提供了异步（Async）、反射式（Reactive）和RxJava2标准的接口。

        RAtomicDouble atomicDouble = redisson.getAtomicDouble("myAtomicDouble");
        atomicDouble.set(2.81);
        atomicDouble.addAndGet(4.11);
        atomicDouble.get();
        // 6.7. 话题（订阅分发）
        // 6.7.1. 模糊话题
        // 6.8. 布隆过滤器（Bloom Filter）
        // 6.8.1. 布隆过滤器数据分片（Sharding）
        // 6.9. 基数估计算法（HyperLogLog）
        // 6.10. 整长型累加器（LongAdder）
        // 6.11. 双精度浮点累加器（DoubleAdder）
        // 6.12. 限流器（RateLimiter）
        return ResultMsg.success();
    }

    @PostMapping("/redisson/reentrantLock")
    public ResultMsg reentrantLock(@RequestBody RedissonDTO redissonDTO) throws InterruptedException {
        RLock lock = RedissonConfig.getRedisson().getLock("anyLock");
        // 最常见的使用方法，会自动续期，但是需要调用unlock()，否则不会释放锁，后续的线程进入会阻塞
        lock.lock();
        // TimeUnit.SECONDS.sleep(35);
        // lock.unlock();

        // 加锁以后10秒钟自动解锁，线程执行完后锁还存在
        // 无需调用unlock方法手动解锁，也可以调用直接解锁
        // lock.lock(10, TimeUnit.SECONDS);

        // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁
        // boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);
        // if (res) {
        //     try {
        //         TimeUnit.SECONDS.sleep(8);
        //     } finally {
        //         lock.unlock();
        //     }
        // }

        // Redisson同时还为分布式锁提供了异步执行的相关方法：
        // RLock lock1 = RedissonDistributedLock.getRedisson().getLock("anyLock");
        // lock1.lockAsync();
        // lock1.lockAsync(10, TimeUnit.SECONDS);
        // Future<Boolean> res1 = lock1.tryLockAsync(100, 10, TimeUnit.SECONDS);
        log.info("我是日志");
        lock.unlock();
        return ResultMsg.success();
    }

    /**
     * 测试失败
     */
    @PostMapping("/redisson/semaphore")
    public ResultMsg semaphore(@RequestBody RedissonDTO redissonDTO) throws InterruptedException {
        RedissonClient redisson = RedissonConfig.getRedisson();
        RSemaphore semaphore = redisson.getSemaphore("semaphore");
        // semaphore.acquire();
        // semaphore.release();
        //或
        // semaphore.acquireAsync();
        semaphore.acquire(23);
        // semaphore.tryAcquire();
        // //或
        // semaphore.tryAcquireAsync();
        // semaphore.tryAcquire(23, TimeUnit.SECONDS);
        // //或
        // semaphore.tryAcquireAsync(23, TimeUnit.SECONDS);
        // semaphore.release(10);
        semaphore.release();
        //或
        semaphore.releaseAsync();
        return ResultMsg.success();
    }

    @PostMapping("/redisson/permitExpirableSemaphore")
    public ResultMsg permitExpirableSemaphore(@RequestBody RedissonDTO redissonDTO) throws InterruptedException {
        RedissonClient redisson = RedissonConfig.getRedisson();
        RPermitExpirableSemaphore semaphore = redisson.getPermitExpirableSemaphore("mySemaphore");
        String permitId = semaphore.acquire();

        // 获取一个信号，有效期只有2秒钟。
        String permitId1 = semaphore.acquire(2, TimeUnit.SECONDS);
        // ...
        semaphore.release(permitId1);
        return ResultMsg.success();
    }

    /**
     * 10.2. 复杂多维对象结构和对象引用的支持
     */
    @PostMapping("/redisson/complexObject")
    public ResultMsg complexObject() throws InterruptedException {
        // edisson突破了Redis数据结构维度的限制，通过一个特殊引用对象的帮助，Redisson允许以任意的组合方式构建多维度的复杂对象结构，实现了对象之间的类似传统数据库里的关联关系。使用范例如下：
        RedissonClient redisson = RedissonConfig.getRedisson();
        RMap<RSet<RList>, RList<RMap>> map = redisson.getMap("myMap");
        RSet<RList> set = redisson.getSet("mySet");
        RList<RMap> list = redisson.getList("myList");

        map.put(set, list);
        // 在特殊引用对象的帮助下，我们甚至可以构建一个循环引用，这是通过普通序列化方式实现不了的。
        set.add(list);
        list.add(map);
        map.expire(20, TimeUnit.SECONDS);
        set.expire(20, TimeUnit.SECONDS);
        list.expire(20, TimeUnit.SECONDS);
        // 可能您已经注意到了，在map包含的元素发生改变以后，我们无需再次“保存/持久”这些对象。因为map对象所记录的并不是序列化以后的值，而是元素对象的引用。这让Redisson提供的对象在使用方法上，与普通Java对象的使用方法一致。从而让Redis成为内存的一部分，而不仅仅是一个储存空间。
        //
        // 以上范例中，一共创建了三个Redis数据结构：一个Redis HASH，一个Redis SET和一个Redis LIST。
        log.info("complexObject");
        return ResultMsg.success();
    }

    /**
     * 10.3. 命令的批量执行
     */
    @PostMapping("/redisson/batchExecutionOfCommands")
    public ResultMsg batchExecutionOfCommands() throws InterruptedException, ExecutionException {
        // 多个连续命令可以通过RBatch对象在一次网络会话请求里合并发送，这样省去了产生多个请求消耗的时间和资源。这在Redis中叫做管道。
        //
        // 用户可以通过以下方式调整通过管道方式发送命令的方式：
        BatchOptions options = BatchOptions.defaults()
                // 指定执行模式
                //
                // ExecutionMode.REDIS_READ_ATOMIC - 所有命令缓存在Redis节点中，以原子性事务的方式执行。
                //
                // ExecutionMode.REDIS_WRITE_ATOMIC - 所有命令缓存在Redis节点中，以原子性事务的方式执行。
                //
                // ExecutionMode.IN_MEMORY - 所有命令缓存在Redisson本机内存中统一发送，但逐一执行（非事务）。默认模式。
                //
                // ExecutionMode.IN_MEMORY_ATOMIC - 所有命令缓存在Redisson本机内存中统一发送，并以原子性事务的方式执行。
                //
                .executionMode(ExecutionMode.IN_MEMORY)
                // 告知Redis不用返回结果（可以减少网络用量）
                .skipResult()
                // 将写入操作同步到从节点
                // 同步到2个从节点，等待时间为1秒钟
                .syncSlaves(2, 1, TimeUnit.SECONDS)
                // 处理结果超时为2秒钟
                .responseTimeout(2, TimeUnit.SECONDS)
                // 命令重试等待间隔时间为2秒钟
                .retryInterval(2, TimeUnit.SECONDS)
                // 命令重试次数。仅适用于未发送成功的命令
                .retryAttempts(4);
        // 使用方式如下：
        RedissonClient redisson = RedissonConfig.getRedisson();
        // RBatch batch = redisson.createBatch(options);
        RBatch batch = redisson.createBatch();
        batch.getMap("test").fastPutAsync("1", "2");
        batch.getMap("test").fastPutAsync("2", "3");
        batch.getMap("test").putAsync("2", "5");
        batch.getAtomicLong("counter").incrementAndGetAsync();
        // batch.getAtomicLongAsync("counter").incrementAndGetAsync();
        // batch.getAtomicLongAsync("counter").incrementAndGetAsync();
        batch.getMap("test").expireAsync(10, TimeUnit.SECONDS);
        batch.getAtomicLong("counter").expireAsync(10, TimeUnit.SECONDS);
        // BatchResult<?> res = batch.execute();
        // 或者
        RFuture<BatchResult<?>> asyncRes = batch.executeAsync();
        BatchResult<?> batchResult = asyncRes.get();
        // 在集群模式下，所有的命令会按各个槽所在的节点，筛选分配到各个节点并同时发送。每个节点返回的结果将会汇总到最终的结果列表里。
        log.info("batchExecutionOfCommands");
        return ResultMsg.success();
    }

    /**
     * 10.4. Redisson事务
     */
    @PostMapping("/redisson/transaction")
    public ResultMsg transaction() throws InterruptedException, ExecutionException {
        // Redisson为RMap、RMapCache、RLocalCachedMap、RSet、RSetCache和RBucket这样的对象提供了具有ACID属性的事务功能。Redisson事务通过分布式锁保证了连续写入的原子性，同时在内部通过操作指令队列实现了Redis原本没有的提交与滚回功能。当提交与滚回遇到问题的时候，将通过org.redisson.transaction.TransactionException告知用户。
        //
        // 目前支持的环境如下： SINGLE, MASTER/SLAVE, SENTINEL, ELASTICACHE REPLICATED, AZURE CACHE, RLEC。
        //
        // Redisson事务支持的事务隔离等级为: READ_COMMITTED，即仅读取提交后的结果。
        //
        // 另见 Spring事务管理器 和本章 XA事务（XA Transactions）。
        //
        // 以下选项可以用来配置事务属性：

        TransactionOptions options = TransactionOptions.defaults()
                // 设置参与本次事务的主节点与其从节点同步的超时时间。
                // 默认值是5秒。
                .syncSlavesTimeout(5, TimeUnit.SECONDS)
                // 处理结果超时。
                // 默认值是3秒。
                .responseTimeout(3, TimeUnit.SECONDS)
                // 命令重试等待间隔时间。仅适用于未发送成功的命令。
                // 默认值是1.5秒。
                .retryInterval(2, TimeUnit.SECONDS)
                // 命令重试次数。仅适用于未发送成功的命令。
                // 默认值是3次。
                .retryAttempts(3)
                // 事务超时时间。如果规定时间内没有提交该事务则自动滚回。
                // 默认值是5秒。
                .timeout(5, TimeUnit.SECONDS);
        // 代码范例：
        RedissonClient redisson = RedissonConfig.getRedisson();
        RTransaction transaction = redisson.createTransaction(TransactionOptions.defaults());

        RMap<String, String> map = transaction.getMap("myMap");
        map.put("1", "2");
        String value = map.get("3");
        RSet<String> set = transaction.getSet("mySet");
        set.add(value);

        try {
            TimeUnit.SECONDS.sleep(4);
            transaction.commit();
        } catch (TransactionException e) {
            // 如果规定时间内没有提交该事务则自动滚回。
            // transaction.rollback();
            log.error("我是异常");
        }
        log.info("transaction");
        return ResultMsg.success();
    }

    /**
     * 10.5. XA事务（XA Transactions）
     */
    @PostMapping("/redisson/XATransactions")
    public ResultMsg XATransactions() throws InterruptedException, ExecutionException {
        // Redisson提供了XAResource标准的实现。该实现可用于JTA事务中。
        //
        // 另见本章Redisson事务和Spring事务管理器。
        //
        // 该功能仅适用于Redisson PRO版本
        //
        // 代码范例：

        // Transaction对象可以从所有兼容JTA接口的事务管理器中获取。
        // Transaction globalTransaction = transactionManager.getTransaction();
        //
        // RXAResource xaResource = redisson.getXAResource();
        // globalTransaction.enlistResource(xaResource);
        //
        // RTransaction transaction = xaResource.getTransaction();
        // RBucket<String> bucket = transaction.getBucket("myBucket");
        // bucket.set("simple");
        // RMap<String, String> map = transaction.getMap("myMap");
        // map.put("myKey", "myValue");
        //
        // transactionManager.commit();
        log.info("XATransactions");
        return ResultMsg.success();
    }

    /**
     * 10.6. 脚本执行
     */
    @PostMapping("/redisson/executeScript")
    public ResultMsg executeScript() throws InterruptedException, ExecutionException {
        RedissonClient redisson = RedissonConfig.getRedisson();
        redisson.getBucket("foo").set("bar");
        String r = redisson.getScript().eval(Mode.READ_ONLY,
                "return redis.call('get', 'foo')", RScript.ReturnType.VALUE);

        // 通过预存的脚本进行同样的操作
        RScript s = redisson.getScript();
        // 首先将脚本保存到所有的Redis主节点
        String res = s.scriptLoad("return redis.call('get', 'foo')");
        // 返回值 res == 282297a0228f48cd3fc6a55de6316f31422f5d17

        // 再通过SHA值调用脚本
        Future<Object> r1 = redisson.getScript().evalShaAsync(Mode.READ_ONLY,
                "282297a0228f48cd3fc6a55de6316f31422f5d17",
                RScript.ReturnType.VALUE, Collections.emptyList());
        log.info("executeScript");
        return ResultMsg.success();
    }

    /**
     * 10.7. 底层Redis客户端
     */
    @PostMapping("/redisson/redisConnection")
    public ResultMsg redisConnection() throws InterruptedException, ExecutionException {
        // Redisson在底层采用了高性能异步非阻塞式Java客户端，它同时支持异步和同步两种通信模式。如果有哪些命令Redisson还没提供支持，也可以直接通过调用底层Redis客户端来实现。Redisson支持的命令在Redis命令和Redisson对象匹配列表里做了详细对比参照。

        // 在使用多个客户端的情况下可以共享同一个EventLoopGroup
        EventLoopGroup group = new NioEventLoopGroup();

        RedisClientConfig config = new RedisClientConfig();
        config.setAddress("redis://localhost:6379") // 或者用rediss://使用加密连接
                // .setPassword("myPassword")
                .setDatabase(0)
                .setClientName("myClient")
                .setGroup(group);

        RedisClient client = RedisClient.create(config);
        RedisConnection conn = client.connect();
        // 或
        RFuture<RedisConnection> connFuture = client.connectAsync();

        Object test = conn.sync(StringCodec.INSTANCE, RedisCommands.SET, "test", 0);
        // 或
        RFuture<Object> test1 = conn.async(StringCodec.INSTANCE, RedisCommands.GET, "test");
        Object o = test1.get();

        // conn.close();
        // 或
        conn.closeAsync();

        client.shutdown();
        // 或
        client.shutdownAsync();
        log.info("redisConnection");
        return ResultMsg.success();
    }

    /**
     * 14.7. Spring事务管理器（Spring Transaction Manager）
     * Redisson为Spring事务（Spring Transaction）提供了基于org.springframework.transaction.PlatformTransactionManager 接口的功能实现。另请查阅事务（Transaction） 章节。
     */
    // @PostMapping("/redisson/redissonTransactionManager")
    // @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.READ_COMMITTED, timeout = 30, readOnly = false, rollbackFor = {
    //         Exception.class, RuntimeException.class})
    // public ResultMsg redissonTransactionManager(@RequestBody RedissonDTO redissonDTO)
    //         throws InterruptedException, ExecutionException {
    //     RTransaction transaction = customRedissonTxManager.getCurrentTransaction();
    //     RMap<String, String> map = transaction.getMap("test1");
    //     map.put("1", "2");
    //     log.info("redissonTransactionManager");
    //     if (Objects.equals("aa", redissonDTO.getName())) {
    //         // 抛异常可以回滚
    //         throw new DiyRuntimeException();
    //     }
    //     return ResultMsg.success();
    // }

    /**
     * 使用Bean名为 customRedissonTxManager
     * <br>todo 无法刷新 customRedissonTxManager，由于缓存transactionManagerCache无法刷新
     *
     * @see org.springframework.transaction.interceptor.TransactionAspectSupport#determineTransactionManager
     * @see TransactionAspectSupport#transactionManagerCache
     * @see TransactionSynchronizationManager#doGetResource(java.lang.Object)
     */
    // @PostMapping("/redisson/redissonTransactionManager1")
    // @Transactional(transactionManager = "customRedissonTxManager", propagation = Propagation.REQUIRED, isolation = Isolation.READ_COMMITTED, timeout = 30, readOnly = false, rollbackFor = {
    //         Exception.class, RuntimeException.class})
    // public ResultMsg redissonTransactionManager1(@RequestBody RedissonDTO redissonDTO)
    //         throws InterruptedException, ExecutionException {
    //     RTransaction transaction = customRedissonTxManager.getCurrentTransaction();
    //     RMap<String, String> map = transaction.getMap("test1");
    //     map.put("1", "2");
    //     log.info("redissonTransactionManager");
    //     if (Objects.equals("aa", redissonDTO.getName())) {
    //         // 抛异常可以回滚
    //         throw new DiyRuntimeException();
    //     }
    //     return ResultMsg.success();
    // }

}
